// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <nebula/types/meta/struct_type.h>
#include <nebula/types/util.h>
#include <nebula/util/vector.h>

namespace nebula {

    constexpr Type::type StructType::type_id;

    class StructType::Impl {
    public:
        explicit Impl(const FieldVector &fields)
                : name_to_index_(internal::create_name_to_index_map(fields)) {}

        const std::unordered_multimap<std::string, int> name_to_index_;
    };

    StructType::StructType(const FieldVector &fields)
            : NestedType(Type::STRUCT), impl_(new Impl(fields)) {
        children_ = fields;
    }

    StructType::~StructType() {}

    std::string StructType::to_string(bool show_metadata) const {
        std::stringstream s;
        s << "struct<";
        for (int i = 0; i < this->num_fields(); ++i) {
            if (i > 0) {
                s << ", ";
            }
            std::shared_ptr<Field> field = this->field(i);
            s << field->to_string(show_metadata);
        }
        s << ">";
        return s.str();
    }

    std::shared_ptr<Field> StructType::get_field_by_name(const std::string &name) const {
        int i = get_field_index(name);
        return i == -1 ? nullptr : children_[i];
    }

    int StructType::get_field_index(const std::string &name) const {
        return internal::lookup_name_index(impl_->name_to_index_, name);
    }

    std::vector<int> StructType::get_all_field_indices(const std::string &name) const {
        std::vector<int> result;
        auto p = impl_->name_to_index_.equal_range(name);
        for (auto it = p.first; it != p.second; ++it) {
            result.push_back(it->second);
        }
        if (result.size() > 1) {
            std::sort(result.begin(), result.end());
        }
        return result;
    }

    FieldVector StructType::get_all_fields_by_name(const std::string &name) const {
        FieldVector result;
        auto p = impl_->name_to_index_.equal_range(name);
        for (auto it = p.first; it != p.second; ++it) {
            result.push_back(children_[it->second]);
        }
        return result;
    }

    turbo::Result<std::shared_ptr<StructType>> StructType::add_field(
            int i, const std::shared_ptr<Field> &field) const {
        if (i < 0 || i > this->num_fields()) {
            return turbo::invalid_argument_error("Invalid column index to add field.");
        }
        return std::make_shared<StructType>(internal::AddVectorElement(children_, i, field));
    }

    turbo::Result<std::shared_ptr<StructType>> StructType::remove_field(int i) const {
        if (i < 0 || i >= this->num_fields()) {
            return turbo::invalid_argument_error("Invalid column index to remove field.");
        }
        return std::make_shared<StructType>(internal::DeleteVectorElement(children_, i));
    }

    turbo::Result<std::shared_ptr<StructType>> StructType::set_field(
            int i, const std::shared_ptr<Field> &field) const {
        if (i < 0 || i >= this->num_fields()) {
            return turbo::invalid_argument_error("Invalid column index to set field.");
        }
        return std::make_shared<StructType>(
                internal::ReplaceVectorElement(children_, i, field));
    }

    std::string StructType::compute_fingerprint() const {
        std::stringstream ss;
        ss << type_id_fingerprint(*this) << "{";
        for (const auto &child: children_) {
            const auto &child_fingerprint = child->fingerprint();
            if (child_fingerprint.empty()) {
                return "";
            }
            ss << child_fingerprint << ";";
        }
        ss << "}";
        return ss.str();
    }

}  // namespace nebula
